package org.ngbx.demo.flink.dd.mokcer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.ngbx.demo.flink.dd.DeviceDataEvent;

import java.util.LinkedList;
import java.util.List;
import java.util.Properties;

public class DeviceDataEventMocker {
    // 延迟：毫秒
    public static final long DELAY = 1000;

    public static void main(String[] args) throws Exception {
        // 创建kafka配置属性
        Properties kafkaProps = createKafkaProperties();

        // 创建Kafka消息的生产者
        KafkaProducer<String, byte[]> producer = new KafkaProducer<>(kafkaProps);

        String topic = "device_data_event";

        List<DeviceDataEvent> eventList = new LinkedList<>();
        eventList.add(new DeviceDataEvent("12w34", 0, null, 0, 1553503182000L));
        // 此处为开机
        eventList.add(new DeviceDataEvent("12w34", 1, 0, 0, 1553503183000L));
        // 此处为开始工作
        eventList.add(new DeviceDataEvent("12w34", 1, 1, 5, 1553503184000L));
        eventList.add(new DeviceDataEvent("12w34", 1, 1, 100, 1553503185000L));
        eventList.add(new DeviceDataEvent("12w34", 1, 1, 150, 1553503186000L));
        // 此处为开始待机
        eventList.add(new DeviceDataEvent("12w34", 1, 0, 150, 1553503187000L));
        // 此处为关机
        eventList.add(new DeviceDataEvent("12w34", 0, null, 0, 1553503188100L));

        eventList.forEach(event -> {
            try {
                ProducerRecord<String, byte[]> record = new DeviceDataEventSerialization(topic)
                        .serialize(event, null);

                producer.send(record);

                System.out.println(event.toString());

                Thread.sleep(DELAY);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    private static Properties createKafkaProperties() {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getCanonicalName());
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
        return kafkaProps;
    }
}
